package com.ken.app.kafka.consumer;

import com.ken.app.entity.logTip;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: Liuk
 * @Mail: leo_work@aliyun.com
 * @Date: 2017/8/18 0018
 * @Desc: kafka 消费者
 */
@Component
@Slf4j
public class KafkaReceiverListener {


    private static final String ZTE_A_TOPIC = "boot.t";
    private static final String SI_U_TOPIC = "S1-U";
    private static final String SIMEE_TOPIC = "DETAIL-CDR-s1mme";

    private static Map<String, Long> totalCountMap = new ConcurrentHashMap<>();
    private static Map<String, Long> currentCountMap = new ConcurrentHashMap<>();
    private static Map<String, Long> startTimeMap = new ConcurrentHashMap<>();

    static {
        totalCountMap.put(ZTE_A_TOPIC, 0L);
        totalCountMap.put(SI_U_TOPIC, 0L);
        totalCountMap.put(SIMEE_TOPIC, 0L);

        currentCountMap.put(ZTE_A_TOPIC, 0L);
        currentCountMap.put(SI_U_TOPIC, 0L);
        currentCountMap.put(SIMEE_TOPIC, 0L);

        startTimeMap.put(ZTE_A_TOPIC, System.currentTimeMillis());
        startTimeMap.put(SI_U_TOPIC, System.currentTimeMillis());
        startTimeMap.put(SIMEE_TOPIC, System.currentTimeMillis());
    }

    /**
     * 日志提示
     */
    private void logTip(String topicName, String message) {
        long countTotal = totalCountMap.get(topicName);
        totalCountMap.put(topicName, ++countTotal);
        long countCurrent = currentCountMap.get(topicName);
        currentCountMap.put(topicName, ++countCurrent);
        log.debug("receive topic='{}' message='{}'", topicName, message);
        System.out.println("接收消息--" + topicName + "-----------" + message);
        long timeMilliscond = System.currentTimeMillis() - startTimeMap.get(topicName);
        if (countTotal % 500000 == 0 || timeMilliscond > 60 * 1000L) {
            log.info("Topic '{}' : {}",
                    topicName, new logTip(countTotal, countCurrent, (double) timeMilliscond / 1000, (int) (countCurrent * 1000 / timeMilliscond)));
            startTimeMap.put(topicName, System.currentTimeMillis());
            currentCountMap.put(topicName, 0L);
        }
    }

    @KafkaListener(topics = ZTE_A_TOPIC)
    public void processZteaMessage(String message) {
        logTip(ZTE_A_TOPIC, message);
    }


    @KafkaListener(topics = SI_U_TOPIC)
    public void processSiuMessage(String message) {
        logTip(SI_U_TOPIC, message);
    }


    @KafkaListener(topics = SIMEE_TOPIC)
    public void processS1mmeMessage(String message) {
        logTip(SIMEE_TOPIC, message);
    }
}
